package com.echo.im.repository.offline.support;

import com.echo.im.repository.offline.OfflineMessageService;
import lombok.extern.slf4j.Slf4j;

import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * 核心点：
 * 1. 增量拉取和增量删除：
 * 新增离线消息：如果redis满了，就把redis中的消息取出来放入mysql，然后删除redis中的消息。然后新增入redis中
 * 拉取消息：每次拉取消息都先从 MySQL 中拉取消息，然后从 Redis 中拉取消息。确保 MySQL 中的消息 ID 小于 Redis 中的消息 ID，以保证全局顺序。
 * 删除消息：在 Redis 和 MySQL 中增量删除消息。使用 lastMsgId 参数来标识需要删除的消息的 ID。
 * 上述的redis和mysql内部分别是局部有序，然后我们的取出redis中最旧的消息放入mysql这个策略有效保证了我们消息的全局有序性
 * 为什么必须要保证这个全局有序性？保证根据lastMsgId删除消息的时候不会出现误删
 * 2. Redis 和 MySQL 一致性
 * 一致性保障：确保 Redis 中的消息 ID 大于 MySQL 中的最大消息 ID，从而保证全局消息顺序。
 * 分布式锁：使用分布式锁来保证对 Redis 和 MySQL 的并发操作的一致性，防止多个线程或服务同时操作导致的数据不一致。
 * 3. 批量操作
 * 批量备份：将 Redis 中超出的消息批量备份到 MySQL，避免每次操作都进行单条消息的处理，提高性能。
 * 批量删除：从 Redis 和 MySQL 中批量删除消息，以减少数据库操作的频率和提高效率。
 *
 * 难点：
 * 全局消息顺序的保证：
 * 问题：Redis 和 MySQL 存储消息时分别有序，但整体上不保证全局顺序。
 * 解决方案：确保 Redis 中的消息 ID 大于 MySQL 中的最大消息 ID。拉取消息时，先从 MySQL 拉取已存在的消息，再从 Redis 拉取较新的消息。
 * 并发一致性问题：
 * 问题：在并发情况下，可能会出现同时对 Redis 和 MySQL 进行读写操作，导致数据不一致。
 * 解决方案：使用分布式锁（DistributedLockManager）来控制对 Redis 和 MySQL 的操作，防止并发操作引起的数据不一致问题。
 * 性能优化：
 * 问题：频繁的 Redis 操作和数据库操作可能会影响性能。
 * 解决方案：实现批量备份和批量删除机制，以减少对 Redis 和 MySQL 的操作频率，提升性能。
 * 亮点：
 * 双重存储方案：
 * Redis 和 MySQL 结合：利用 Redis 作为快速访问的缓存层，MySQL 作为持久化存储，结合两者的优点，实现高效的离线消息管理。
 * 分布式锁的使用：
 * 确保一致性：通过 DistributedLockManager 类实现分布式锁，防止并发操作导致的消息不一致问题。
 * 批量处理：
 * 性能优化：在备份和删除消息时，使用批量处理机制提高性能，避免频繁的数据库操作。
 * 哪些地方需要加分布式锁？
 * 1. 增量备份消息到 MySQL：
 * 防止重复备份、确保数据一致性：在并发情况下，如果多个线程或服务同时尝试备份 Redis 中的消息到 MySQL，可能会导致重复备份或数据不一致。使用分布式锁可以确保只有一个线程或服务在特定时间内执行备份操作。
 * 2. 增量删除消息
 * 在高并发环境下，多个线程或服务可能同时尝试删除 Redis 和 MySQL 中的消息。使用分布式锁可以确保只有一个线程执行删除操作，避免数据遗漏或重复删除。
 * 3.
 * @author echo
 * @date 2024-08-17
 */
@Slf4j
public abstract class AbstractOfflineService<T> implements OfflineMessageService<T> {

    private static final String DISTRIBUTED_LOCK_SUFFIX = ":lock";

    private static final long LOCK_TIMEOUT_SECONDS = 10;

    private final DistributedLockManager lockManager;

    private final RedisOfflineStorage redisStorage;
    /**
     * 构造方法，初始化分布式锁管理器
     * @param lockManager 分布式锁管理器
     */
    public AbstractOfflineService(DistributedLockManager lockManager, RedisOfflineStorage redisStorage) {
        this.lockManager = lockManager;
        this.redisStorage = redisStorage;
    }

    /**
     * 获取Redis键前缀，由子类实现
     * @return Redis键前缀
     */
    protected abstract String getRedisKeyPrefix();

    /**
     * 为指定用户获取分布式锁，返回获取结果
     * @param userId 用户ID
     * @param sessionUuid 会话UUID
     * @return 获取锁结果，true表示成功，false表示失败
     */
    private boolean acquireUserLock(Long userId, String sessionUuid) {
        String lockKey = getRedisKeyPrefix() + userId + DISTRIBUTED_LOCK_SUFFIX;
        try {
            return lockManager.acquireLockWithUuid(lockKey, sessionUuid, LOCK_TIMEOUT_SECONDS, TimeUnit.SECONDS);
        } catch (Exception e) {
            log.error("获取用户 {} 的锁时发生错误", userId, e);
            return false;
        }
    }

    /**
     * 释放指定用户的分布式锁
     * @param userId 用户ID
     * @param sessionUuid 会话UUID
     */
    private void releaseUserLock(Long userId, String sessionUuid) {
        String lockKey = getRedisKeyPrefix() + userId + DISTRIBUTED_LOCK_SUFFIX;
        lockManager.releaseLockWithUuid(lockKey, sessionUuid);
    }

    /**
     * 从存储中获取并删除离线消息
     * @param userId 用户ID
     * @param lastFetchedMessageId 上次获取的最后一条消息ID，用于分页查询
     * @param fetchLimit 本次获取的消息数量限制
     * @param sessionUuid 会话UUID
     * @return 离线消息列表
     */
    @Override
    public List<T> fetchAndDeleteOfflineMessages(Long userId, Long lastFetchedMessageId, int fetchLimit, String sessionUuid) {
        String lockKey = getRedisKeyPrefix() + userId + DISTRIBUTED_LOCK_SUFFIX;

        if (Objects.isNull(lastFetchedMessageId)) {
            if (!acquireUserLock(userId, sessionUuid)) {
                throw new IllegalStateException("无法获取用户的锁，可能被其他会话占用");
            }
        } else {
            if (!lockManager.isLockOwner(lockKey, sessionUuid)) {
                throw new IllegalStateException("用户未持有锁或锁已被其他会话占用");
            }
            lockManager.renewLock(lockKey, LOCK_TIMEOUT_SECONDS, TimeUnit.SECONDS);
        }

        try {
            if (Objects.isNull(lastFetchedMessageId)) {
                lastFetchedMessageId = 0L;
            }

            List<T> messages = fetchAndDeleteMessagesFromStorage(userId, lastFetchedMessageId, fetchLimit);

            if (messages.isEmpty()) {
                releaseUserLock(userId, sessionUuid);
            }

            return messages;
        } catch (Exception e) {
            log.error("获取并删除用户 {} 的离线消息时发生错误", userId, e);
            return Collections.emptyList();
        }
    }

    /**
     * 从存储中获取并删除离线消息，由子类实现
     * @param userId 用户ID
     * @param lastFetchedMessageId 上次获取的最后一条消息ID
     * @param fetchLimit 本次获取的消息数量限制
     * @return 离线消息列表
     */
    protected abstract List<T> fetchAndDeleteMessagesFromStorage(Long userId, Long lastFetchedMessageId, int fetchLimit);

    /**
     * 向指定用户添加离线消息
     * @param userId 用户ID
     * @param msgId 消息ID
     */
    @Override
    public void addOfflineMessage(Long userId, Long msgId) {
        String redisKey = getRedisKeyPrefix()+userId;
        doStoreMessage(userId, redisKey, msgId);
    }

    /**
     * 目前只需要查Redis中的，mysql不用管
     * @param userId 用户Id
     * @return
     */
    @Override
    public Long getOfflineMessageCount(Long userId) {
        Long count=redisStorage.getMessageCount(getRedisKeyPrefix()+userId);
        return count;
    }


    /**
     * 将消息存储到Redis中，由子类实现
     * @param userId 用户ID
     * @param redisKey Redis键
     * @param msgId 消息ID
     */
    protected abstract void doStoreMessage(Long userId, String redisKey, Long msgId);
}
